package climspace.hadoop;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;

// filter station in the map
// filter dates in reduce
// job can be submit asynchronously or synchronously

@SuppressWarnings("deprecation")
public class GetHistoricalData {
	private String hostName;
	private String hdfsPort;
	private String mapredPort;
	
	public GetHistoricalData(String hostName, String hdfsPort, String mapredPort) {
		this.hostName = hostName;
		this.hdfsPort = hdfsPort;
		this.mapredPort = mapredPort;
	}
	
	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		String stationName = "*";
		String fromDate = "-9999999";
		String toDate = "99999999";
		
		public void configure(JobConf conf) {
			String stationName = conf.get("ClimSpace.Filter.StationName");
			if (stationName.equals("*")==false) {
				this.stationName = stationName;
			}
			
			String fromDate = conf.get("ClimSpace.Filter.FromDate");
			if ( fromDate.equals("*")==false) {
				this.fromDate = fromDate;
			}
			
			String toDate = conf.get("ClimSpace.Filter.ToDate");
			if (toDate.equals("*")==false) {
				this.toDate = toDate;
			}
		}
		
		public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// map will filter on date
			
			String line = value.toString();
			int intFromDate = Integer.parseInt(fromDate);
			int intToDate = Integer.parseInt(toDate);
			
			try {
				String measureDate = (line.substring(9, 17)).trim();
				int currentDate = Integer.parseInt(measureDate);

				if (intFromDate <= currentDate && intToDate >= currentDate) {
					String station = (line.substring(0,7)).trim();
					if (this.stationName.equals("*")) { 
						output.collect(value, new Text());
					} else if (this.stationName.equals(station)) {
						output.collect(value, new Text());
					}
				}
			} catch (NumberFormatException e) {
				System.out.println("Skipping non data lines");
			} catch (StringIndexOutOfBoundsException e) {
				System.out.println("Skipping non data lines");
			}
		}
	}

	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {	
		public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output,
				Reporter reporter) throws IOException {
			output.collect(key, null);
		}
	}
	
	/**
	 * @param args[0] input folder on HDFS
	 * @param args[1] output folder on HDFS
	 * @param args[2] station id
	 * @throws Exception
	 */
	public int runAsyncJob(String inputPath, String outputPath, String stationName, String fromDate, String toDate){
		JobConf conf = new JobConf(GetHistoricalData.class);
		conf.setJobName("GetHistoricalData");
		
//		conf.set("fs.default.name", "hdfs://localhost:9000/");
//		conf.set("mapred.job.tracker", "localhost:9001");

		conf.set("fs.default.name", "hdfs://" + hostName + ":" + hdfsPort + "/");
		conf.set("mapred.job.tracker", hostName + ":" + mapredPort);
		
		conf.set("ClimSpace.Filter.StationName", stationName);
		conf.set("ClimSpace.Filter.FromDate", fromDate);
		conf.set("ClimSpace.Filter.ToDate", toDate);

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapperClass(Map.class);
//		conf.setCombinerClass(Reduce.class);
		conf.setReducerClass(Reduce.class);
//		conf.setJar("climspace_hadoop.jar");

		conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
		conf.setOutputFormat(org.apache.hadoop.mapred.TextOutputFormat.class);

		Path output = new Path(outputPath);
		FileSystem fs;
		try {
			fs = FileSystem.get(conf);
			fs.delete(output, true);
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		
		org.apache.hadoop.mapred.FileInputFormat.setInputPaths(conf, new Path(inputPath));
		org.apache.hadoop.mapred.FileOutputFormat.setOutputPath(conf, output);
		
		// JobClient.runJob(conf);
		
		JobClient jc;
		int jobID = 0;
		
		try {
			jc = new JobClient(conf);
			RunningJob job = jc.submitJob(conf);
			jobID = job.getID().getId();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		return jobID;
	}
	
	public static void main(String[] args) throws Exception {
		if (args.length != 5)
			throw new RuntimeException("Incorrent number of parameter");
		GetHistoricalData  wc = new GetHistoricalData("localhost", "9000", "9001");
		wc.runAsyncJob(args[0], args[1], args[2], args[3], args[4]);
	}
}
